package cn.sheep.dolphin.tag

import cn.sheep.dolphin.common.{DolphinAppComm, Jedispools}
import cn.sheep.dolphin.utils.TagsUtils
import org.apache.spark.sql.SQLContext

/**
  * 用户的特征数据提取(用户画像)
  * author: old sheep
  * QQ: 64341393 
  * Created 2018/12/2
  */
object UserTags {

	def main(args: Array[String]): Unit = {

		if (args.size != 3) {
			println(
				"""
				  |Usage: cn.sheep.dolphin.tag.UserTags <parquetInputPath, tagOutputPath, stopwordPath>
				""".stripMargin)
			sys.exit(-1)
		}
		// 接受参数
		val Array(parquetInputPath, stopwordPath, tagOutputPath) = args

		val sc = DolphinAppComm.createSparkContext("用户的特征数据提取")
		// 广播停用词字典数据
		val stopWords = sc.textFile(stopwordPath).map((_, null)).collectAsMap()
		val stopwordBT = sc.broadcast(stopWords)

		// 读取parquet文件
		val sqlc = new SQLContext(sc)
		val dataFrame = sqlc.read.parquet(parquetInputPath)

		// 过滤掉非法的数据
		val filtered = dataFrame.filter(
			"""
			  |imei!='' or idfa!='' or mac!='' or androidid!='' or openudid!='' or
			  |imeimd5!='' or idfamd5!='' or macmd5!='' or androididmd5!='' or openudidmd5!='' or
			  |imeisha1!='' or idfasha1!='' or macsha1!='' or androididsha1!='' or openudidsha1!=''
			""".stripMargin)

		filtered.mapPartitions(partition => {
			// 从连接池中获取一个连接
			val jedis = Jedispools.getJedis()

			val partitionResult = partition.map(row => {
				// 获取用户的某一个不为空的id
				val userId = TagsUtils.getOneUserId(row)
				(userId, (Tags4Device.makeTags(row, jedis) ++ Tags4Ads.makeTags(row, stopwordBT)).toList)
			})
			jedis.close()
			partitionResult
		})
			.reduceByKey((list1, list2) => (list1 ++ list2).groupBy(_._1).mapValues(_.size).toList)
			.map(tp => tp._1 + "\t" + tp._2.map { case (k, v) => k + ":" + v }.mkString(","))
			.saveAsTextFile(tagOutputPath)


		sc.stop()

	}

}
